package org.example.insert_into;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonInsertMultipleTable {
    public static void main(String[] args) {
        // 创建执行环境
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 创建 Paimon Catalog
        System.out.println("创建 Paimon Catalog:");
        String createCatalogSql = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ")";
        System.out.println(createCatalogSql);
        tableEnv.executeSql(createCatalogSql);

        // 创建 DataGen 源表
        System.out.println("\n创建 DataGen 源表:");
        String createDataGenSql = "CREATE TABLE behavior_gen (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    dt AS '2025-01-12',\n" +
                "    eventtime TIMESTAMP_LTZ(3),\n" +
                "    -- 使用 MOD 函数生成 0 或 1\n" +
                "    table_name AS CASE MOD(CAST(behavior AS INT), 2)\n" +
                "                    WHEN 0 THEN 'user_behavior1'\n" +
                "                    ELSE 'user_behavior2'\n" +
                "                END\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '200'\n" +
                ")";
        System.out.println(createDataGenSql);
        tableEnv.executeSql(createDataGenSql);

        // 创建 Paimon 目标表1
        System.out.println("\n创建 Paimon 表1:");
        String createTable1Sql = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior1` (\n" +
                "    user_id STRING COMMENT '用户ID',\n" +
                "    item_id STRING COMMENT '商品ID',\n" +
                "    behavior BIGINT COMMENT '行为类型',\n" +
                "    dt STRING COMMENT '日期',\n" +
                "    eventtime TIMESTAMP_LTZ(3) COMMENT '事件时间',\n" +
                "    WATERMARK FOR eventtime AS eventtime - INTERVAL '5' SECOND,\n" +
                "    PRIMARY KEY (dt, user_id) NOT ENFORCED\n" +
                ") PARTITIONED BY (dt)\n" +
                "WITH (\n" +
                "    'changelog-producer' = 'lookup'\n" +
                ")";
        System.out.println(createTable1Sql);
        tableEnv.executeSql(createTable1Sql);

        // 创建 Paimon 目标表2
        System.out.println("\n创建 Paimon 表2:");
        String createTable2Sql = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior2` (\n" +
                "    user_id STRING COMMENT '用户ID',\n" +
                "    item_id STRING COMMENT '商品ID',\n" +
                "    behavior BIGINT COMMENT '行为类型',\n" +
                "    dt STRING COMMENT '日期',\n" +
                "    eventtime TIMESTAMP_LTZ(3) COMMENT '事件时间',\n" +
                "    WATERMARK FOR eventtime AS eventtime - INTERVAL '5' SECOND,\n" +
                "    PRIMARY KEY (dt, user_id) NOT ENFORCED\n" +
                ") PARTITIONED BY (dt)\n" +
                "WITH (\n" +
                "    'changelog-producer' = 'lookup'\n" +
                ")";
        System.out.println(createTable2Sql);
        tableEnv.executeSql(createTable2Sql);

        // 使用 STATEMENT SET 同时执行多个 INSERT
        System.out.println("\n执行数据分发 SQL:");
        String statementSet = "EXECUTE STATEMENT SET\n" +
                "BEGIN\n" +
                "    INSERT INTO `paimon`.`default`.`user_behavior1`\n" +
                "    SELECT user_id, item_id, behavior, dt, eventtime\n" +
                "    FROM behavior_gen\n" +
                "    WHERE table_name = 'user_behavior1';\n" +
                "\n" +
                "    INSERT INTO `paimon`.`default`.`user_behavior2`\n" +
                "    SELECT user_id, item_id, behavior, dt, eventtime\n" +
                "    FROM behavior_gen\n" +
                "    WHERE table_name = 'user_behavior2';\n" +
                "END";
        System.out.println(statementSet);
        tableEnv.executeSql(statementSet);
    }
}
